/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package hadoopStuff;

import java.io.IOException;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;
import java.io.*;

import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.compress.*;

/**
 *
 * @author hubert
 */
public class idfCalculator {
    public static JobConf conf = new JobConf(idfCalculator.class);

    public static class Map extends MapReduceBase implements Mapper<LongWritable, Text, Text, Text> {
        //private final static IntWritable one = new IntWritable(1);
        
        
        private Text articleNm = new Text();
        private Text word = new Text();

        public void map(LongWritable key, Text value, OutputCollector<Text, Text> output, Reporter reporter) throws IOException {
            String rawArticleNm = "";
            if(conf.get("map.input.file") != null){
                rawArticleNm = conf.get("map.input.file");
            }
            String line = value.toString();
            StringTokenizer tokenizer = new StringTokenizer(line);
            while (tokenizer.hasMoreTokens()) {
                word.set(tokenizer.nextToken());
                output.collect(word, articleNm);
            }
        }
    }

    public static class Reduce extends MapReduceBase implements Reducer<Text, Text, Text, IntWritable> {

        public void reduce(Text key, Iterator<Text> values, OutputCollector<Text, IntWritable> output, Reporter reporter) throws IOException {
            int sum = 0;
            String prevArticle = "";
            while (values.hasNext()) {
                if (values.next().toString().compareTo(prevArticle) != 0) {
                    sum = sum + 1;
                }

                //sum += values.next().get();
            }
            output.collect(key, new IntWritable(sum));
        }
//        public static class inputFormat extends org.apache.hadoop.mapred.FileInputFormat<Text, Text> implements JobConfigurable {
//
//            private CompressionCodecFactory compressionCodecs = null;
//
//            public void configure(JobConf conf) {
//                compressionCodecs = new CompressionCodecFactory(conf);
//            }
//
//            protected boolean isSplitable(FileSystem fs, Path file) {
//                return compressionCodecs.getCodec(file) == null;
//            }
//
//            public RecordReader<Text, Text> getRecordReader(
//                    InputSplit genericSplit, JobConf job,
//                    Reporter reporter)
//                    throws IOException {
//
//                reporter.setStatus(genericSplit.toString());
//                return new LineRecordReader(job, (FileSplit) genericSplit);
//            }
//        }
    }

    public static void main(String[] args) throws Exception {
        
        conf.setJobName("dfCalculator");

        conf.setOutputKeyClass(Text.class);
        conf.setOutputValueClass(IntWritable.class);

        conf.setMapperClass(Map.class);
        conf.setCombinerClass(Reduce.class);
        conf.setReducerClass(Reduce.class);

        conf.setInputFormat(TextInputFormat.class);
        conf.setOutputFormat(TextOutputFormat.class);
        //conf.setInputPath(new Path(args[0]));
        //conf.setOutputPath(new Path(args[1]));

        FileInputFormat.setInputPaths(conf, new Path(args[0]));
        FileOutputFormat.setOutputPath(conf, new Path(args[1]));

        JobClient.runJob(conf);
    }
}
